#include "common/log/log.h"
#include "sql/operator/aggre_physical_operator.h"
#include "storage/record/record.h"
#include "storage/table/table.h"
#include "sql/parser/value.h"
#include "sql/parser/parse_defs.h"

RC AggrePhysicalOperator::open(Trx *trx)
{
  if (children_.empty()) {
    return RC::SUCCESS;
  }

  for(int i=0;i<this->children_.size();i++){
    PhysicalOperator *child = children_[i].get();
    RC rc = child->open(trx);
    if (rc != RC::SUCCESS) {
        LOG_WARN("failed to open child operator: %s", strrc(rc));
        return rc;
    }
  }
  
  return RC::SUCCESS;
}

RC do_aggre(Aggre &aggre ,PhysicalOperator* child, Value &value);

RC AggrePhysicalOperator::next()
{
  if (children_.empty()||this->is_read_==1) {
    return RC::RECORD_EOF;
  }
  
  for(int i=0;i<this->all_aggres_.size();i++){
    Aggre aggre = this->all_aggres_[i];
    PhysicalOperator *child = children_[i].get();
    Value value;
    RC rc=do_aggre(aggre,child,value);
    if(rc!=RC::SUCCESS){
      return rc;
    }
    values_.push_back(value);
  }
  this->is_read_ = 1;

  return RC::SUCCESS;
}

RC AggrePhysicalOperator::close()
{
  for(int i=0;i<this->children_.size();i++){
    PhysicalOperator *child = children_[0].get();
    RC rc = child->close();
    if (rc != RC::SUCCESS) {
        LOG_WARN("failed to close child operator: %s", strrc(rc));
        return rc;
    }
  }
  return RC::SUCCESS;
}

Tuple *AggrePhysicalOperator::current_tuple()
{
  this->tuple_.set_cells(values_);
  return &tuple_;
}

RC do_aggre(Aggre &aggre ,PhysicalOperator* child, Value &value)
{
  Value max_value;
  Value min_value;
  switch (aggre.get_flag())
  {
  case AGGREGATE_COUNT:{
    int count=0;
    while(true){
      RC rc = child->next();
      if(rc==RC::RECORD_EOF){
        break;
      }
      if(rc!=RC::SUCCESS){
        return rc;
      }
      count++;
    }
    value.set_int(count);
  }
    break;
  case AGGREGATE_SUM:{
    int sum=0;
    while(true){
      RC rc = child->next();
      if(rc==RC::RECORD_EOF){
        break;
      }
      if(rc!=RC::SUCCESS){
        return rc;
      }
      Tuple *tuple = child->current_tuple();
      Value cell;
      tuple->cell_at(0,cell);
      sum+=cell.get_float();
    }
    value.set_float(sum);
  }
    break;
  case AGGREGATE_AVG:{
    float avg_sum=0.0;
    float avg_count=0.0;
    while(true){
      RC rc = child->next();
      if(rc==RC::RECORD_EOF){
        break;
      }
      if(rc!=RC::SUCCESS){
        return rc;
      }
      Tuple *tuple = child->current_tuple();
      Value cell;
      tuple->cell_at(0,cell);
      avg_sum+=cell.get_float();
      avg_count++;
    }
    value.set_float(avg_sum/(float)avg_count);
  }
    break;
  case AGGREGATE_MAX:{
    int max=0;
    while(true){
      RC rc = child->next();
      if(rc==RC::RECORD_EOF){
        break;
      }
      if(rc!=RC::SUCCESS){
        return rc;
      }
      Tuple *tuple = child->current_tuple();
      Value cell;
      tuple->cell_at(0,cell);
      
      if(max==0||cell.compare(max_value)>0){
        max_value.set_value(cell);
        max=1;
      }
    }
    value.set_value(max_value);
  }
    break;
  case AGGREGATE_MIN:{
    int min=0;
    while(true){
      RC rc = child->next();
      if(rc==RC::RECORD_EOF){
        break;
      }
      if(rc!=RC::SUCCESS){
        return rc;
      }
      Tuple *tuple = child->current_tuple();
      Value cell;
      tuple->cell_at(0,cell);
      if(min==0||cell.compare(min_value)<0){
        min_value.set_value(cell);
        min=1;
      }
    }
    value.set_value(min_value);
  }
    break;
  }
  return RC::SUCCESS;
}

